package com.ruoyi.project.tugboat.task;

import com.ruoyi.project.tugboat.domain.TMqttMsg;
import com.ruoyi.project.tugboat.mapper.TMqttMsgMapper;
import com.ruoyi.project.tugboat.service.ITMqttMsgService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;


import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 定时轮巡数据库自动转换MQTT报文
 */
@Component("MqttTask")
public class MqttTask {
    @Autowired
    private TMqttMsgMapper tMqttMsgMapper;
    @Autowired
    private ITMqttMsgService itMqttMsgService;
    
    // MQTT消息自动转换定时任务
    public void autoConvertMqttMsg(){
        List<TMqttMsg> tMqttMsgList = tMqttMsgMapper.selectAllNotConverted();
        System.out.println(tMqttMsgList.size());
        if (tMqttMsgList.size()!=0){
            // SQL语句中的in (...)最多只能有1000个元素，所以要分批次处理
            for (int i=0;i<tMqttMsgList.size();i+=1000){
                List<TMqttMsg> subTMqttMsgList;
                if (i+1000<tMqttMsgList.size()){
                    subTMqttMsgList = tMqttMsgList.subList(i,i+1000);
                }else{
                    subTMqttMsgList = tMqttMsgList.subList(i,tMqttMsgList.size());
                }
                // 使用线程安全的队列并行处理
                ConcurrentLinkedQueue<TMqttMsg> concurrentLinkedQueue = new ConcurrentLinkedQueue<>(subTMqttMsgList);
                ExecutorService executorService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
                concurrentLinkedQueue.parallelStream().forEach(tMqttMsg -> {
                    try {
                        // 转换报文
                        itMqttMsgService.convertOriginalTMqttMsg(tMqttMsg);
                        // 转换成功则更新convertFlag为1
                        tMqttMsgMapper.updateConvertFlag(tMqttMsg);
                    } catch (Exception e) {
                        // 转换失败则更新convertFlag为2并记录转换失败原因
                        tMqttMsgMapper.updateConvertFlagAndFailReason(tMqttMsg,e.getMessage());
                    }
                });
                // 关闭队列
                executorService.shutdown();
            }
        }
    }
    
    
}
